package gnss_plugs

/***
  实时数据流写入文件库
*/

import (
	"dgo/goutils/gnss"
	"dgo/goutils/utils"
	"fmt"
	"math"
	"sync"
	"sync/atomic"
	"time"
)

/*
**

	实时数据流写入类
	 推荐使用
*/
type GnssRinexObs3xWriterRt struct {
	__writing_flag int32 // 正在写入

	gnssRinexObs3xWriterBase
	recv_epoch_n int32 // 接收到的历元数, 不包含过滤的历元
	Id           string
	MountPoint   string

	msmInput *gnss.RtcmMsmEpochInput

	// 当前历元缓存块
	curEpochCacheRec *gnssRinexObs3xWriterEpochCacheRec

	cacheEpochReclst *utils.SyncCycleList // 等待写入缓存块
	cacheEpochNum    int                  // 设定缓存历元数量, 满了后, 推入待写入列表, 比如一秒1个, 设定60个, 则一分钟会投递到待写入列表, 默认60 * 5 最小5

	// 设定的最小写入间隔, 超过该时间 就进行文件写入
	//   0: 忽略处理
	WriteInterval time.Duration

	firstRecvRtcmT time.Time // 第一次接收到Rtcm数据的时间, 系统时间

	FilterNaviSys string // 只存储匹配的系统  GREC
	FixInterval   int64  // 固定采样间隔 单位：秒  5s,  mod(t, 5) = 0

	closeW sync.WaitGroup

	// 检测系统重入bug
	lastgid          uint64
	checkworkingflag int32
	lastdebuginfo    string
	DebuginfoFunc    func() string
}

func NewGnssRinexObs3xWriterRt() *GnssRinexObs3xWriterRt {
	rval := &GnssRinexObs3xWriterRt{
		Id: utils.RandKeyString(4),

		cacheEpochNum: 60 * 5,
	}
	rval.AppendIfFileExist = 1
	rval.Filefmt = ""
	rval.writeCacheBlockSize = utils.SIZE_MB * 2
	return rval
}

/*
**

	配置缓存数量
*/
func (this *GnssRinexObs3xWriterRt) ConfigCacheEpochNum(v int) {
	this.cacheEpochNum = v
}

func (this *GnssRinexObs3xWriterRt) GetFirstRecvRtcmT() time.Time {
	return this.firstRecvRtcmT
}

func (this *GnssRinexObs3xWriterRt) GetLastSuccWriteFileT() time.Time {
	return this.lastWriteFileSuccT
}

func (this *GnssRinexObs3xWriterRt) GetWriteFileSize() int64 {
	return this.writefile_size
}

func (this *GnssRinexObs3xWriterRt) GetLastWriteWarning() string {
	return this.lastWringWaring
}

func (this *GnssRinexObs3xWriterRt) GetRecvEpochNum() int32 {
	return this.recv_epoch_n
}

func (this *GnssRinexObs3xWriterRt) GetLastWriteInfo() string {
	return this.lastWriteInfo
}

func (this *GnssRinexObs3xWriterRt) GetLastWriteStep() string {
	return this.lastWriteStep
}

/*
*

	触发一次写入
	单线程执行
*/
func (this *GnssRinexObs3xWriterRt) writeCache2File() {
	defer utils.PanicHandler()
	lst := this.cacheEpochReclst
	if lst == nil {
		return
	}

	if lst.Count() == 0 {
		return
	}

	if !atomic.CompareAndSwapInt32(&this.__writing_flag, 0, 1) {
		return
	}
	this.lastWriteCacheT = time.Now()
	this.closeW.Add(1)
	defer func() {
		this.lastWriteStep = fmt.Sprintf("[%s]991, flag:[%d]", utils.NowString(), this.__writing_flag)
		this.closeW.Done()
		this.__writing_flag = 0
		this.lastWriteStep = fmt.Sprintf("[%s]993, flag:[%d]", utils.NowString(), this.__writing_flag)
	}()
	for {
		this.lastWriteStep = fmt.Sprintf("[%s]0", utils.NowString())
		if ok, rec := lst.PeekFirst(); ok {
			this.lastWriteStep = fmt.Sprintf("[%s]1", utils.NowString())
			v := utils.GetFirstFromVars(rec)
			if v == nil {
				this.lastWringWaring = fmt.Sprintf("[%s]cacheBlock中第一个数据非法, 为nil", utils.NowString())
				lst.RemoveFirst() // 非法的数据
				continue
			}
			this.lastWriteStep = fmt.Sprintf("[%s]2", utils.NowString())
			cacheRec := v.(*gnssRinexObs3xWriterEpochCacheRec)
			w := this.checkWriter(cacheRec) // 获取写入器
			if w == nil || w.IO == nil {
				break
			}
			if w.write_size == 0 || w.rinexHeader == nil {
				this.lastWriteStep = fmt.Sprintf("[%s]3", utils.NowString())
				w.rinexHeader = this.checkRinexHeader(cacheRec.ObsFileName)
			}
			if w.rinexHeader == nil {
				break

			}

			this.lastWriteStep = fmt.Sprintf("[%s]4", utils.NowString())
			t1 := time.Now()
			s1 := this.writefile_size
			if this.tryWriteBlock(w, cacheRec) { // 处理完成
				lst.RemoveFirst()
				{
					// 处理写入统计信息
					t2 := time.Now().Sub(t1)
					n1 := this.writefile_size - s1
					if t2.Seconds() == 0 {
						this.lastWriteInfo = fmt.Sprintf("[%s] size:%s, t:%.2f (s)", utils.NowString(), utils.HumanFilesize(n1), t2.Seconds())
					} else {
						this.lastWriteInfo = fmt.Sprintf("[%s] size:%s, t:%.2f (s), speed:%s/s", utils.NowString(), utils.HumanFilesize(n1), t2.Seconds(), utils.HumanFilesize(int64(float64(n1)/t2.Seconds())))
					}
				}
			} else {
				this.lastWringWaring = fmt.Sprintf("[%s]tryWriteBlock写入失败", utils.NowString())
				break
			}
		} else {
			this.lastWriteStep = fmt.Sprintf("[%s]981", utils.NowString())
			break
		}
	}
}

func (this *GnssRinexObs3xWriterRt) checkPostCurrentCache() bool {
	rec := this.curEpochCacheRec
	if rec != nil {
		if this.cacheEpochReclst == nil {
			this.cacheEpochReclst = utils.NewSyncCycleList()
			this.cacheEpochReclst.ConfigMaxSize(5)
		}
		this.cacheEpochReclst.Push(this.curEpochCacheRec)
		this.curEpochCacheRec = nil
		return true
	}
	return false
}

func (this *GnssRinexObs3xWriterRt) GetStatusString() string {
	lst := this.cacheEpochReclst
	if lst == nil {
		return fmt.Sprintf("epoch-cache-block-num:%d, recv-epoch-n:%d ", 0, this.recv_epoch_n)
	}
	return fmt.Sprintf("epoch-cache-block-num:%d, recv-epoch-n:%d ", lst.Count(), this.recv_epoch_n)
}

/*
**

	将历元投入当前缓存, 并判断是否需要开一个新的缓存记录
*/
func (this *GnssRinexObs3xWriterRt) checkInputEpoch(epoch *gnss.RtcmMsmEpoch) {
	rec := this.curEpochCacheRec
	if rec != nil {
		strFileN := utils.ParseTimeFmt(this.Filefmt, epoch.EpochTime)
		if strFileN != rec.ObsFileName {
			if this.checkPostCurrentCache() {
				this.requestWrite2File()
			}
		}

	}

	if rec == nil {
		this.curEpochCacheRec = newEpochCacheRec(this.cacheEpochNum)
		rec = this.curEpochCacheRec
		rec.ObsFileName = utils.ParseTimeFmt(this.Filefmt, epoch.EpochTime)
		rec.BeginObsTime = epoch.EpochTime
	}

	atomic.AddInt32(&this.recv_epoch_n, 1)
	rec.epochList.Push(epoch)
	if rec.epochList.Count() == int32(this.cacheEpochNum) { // 队列满了
		if this.checkPostCurrentCache() {
			this.requestWrite2File()
		}
	}
	return
}

func (this *GnssRinexObs3xWriterRt) Stop() {
	this.NotifyDone()
	time.Sleep(time.Millisecond * 100)
	this.closeW.Wait()
}

func (this *GnssRinexObs3xWriterRt) requestWrite2File() {
	if this.cacheEpochReclst == nil || this.cacheEpochReclst.Count() == 0 {
		return
	}
	utils.DefaultWorkers().PostTaskFunc(this, func() {
		this.writeCache2File()
	})
}

func (this *GnssRinexObs3xWriterRt) NotifyDone() {
	if this.msmInput != nil {
		this.msmInput.Close()
	}
	this.checkPostCurrentCache()
	this.requestWrite2File()
}

func (this *GnssRinexObs3xWriterRt) OnRecvGnssRecCallBack(sender interface{}, channelid string, subscribeid interface{}, data interface{}, tag interface{}, isBreak *bool) bool {
	if atomic.CompareAndSwapInt32(&this.checkworkingflag, 0, 1) {
		this.lastgid = utils.GetCurrentGoRoutineID()
		if this.DebuginfoFunc != nil {
			this.lastdebuginfo = this.DebuginfoFunc()
		}
		defer func() {
			this.checkworkingflag = 0
		}()
		if data == nil {
			this.NotifyDone()
			return true
		}
		rtcmRec, ok := data.(*gnss.GnssRtcmData)
		if ok {
			this.OnRecvGnssDataRec(sender, rtcmRec)
		}
		return true
	} else {
		lastid := this.lastgid
		curr_gid := utils.GetCurrentGoRoutineID()
		debuginfo := "[需要设定debuginfofunc,进行获取信息]"
		if this.DebuginfoFunc != nil {
			debuginfo = this.DebuginfoFunc()
		}
		str := fmt.Sprintf("[并发重入]%s(%s), current_gid:%d\r\n debuginfo:%s\r\n "+"==============\r\n"+
			"lastgid:%d\r\n lastdebuginfo:%s\r\n", this.Id, this.MountPoint,
			curr_gid, debuginfo,
			lastid, this.lastdebuginfo)
		utils.Log.WarnTypef("GNSS-BUG", str)
		return false
	}

}

/*
*

	返回false 表示需要过滤掉的数据
*/
func (this *GnssRinexObs3xWriterRt) checkFilterMsm(obs *gnss.RtcmMsm) bool {
	if this.FixInterval > 0 {
		f_msec := float64(obs.MsmTime.UnixNano()) / 1e6 // 毫秒 采样率有 10Hz的数据
		i_msec := int64(math.Floor(f_msec + 0.1))
		v := i_msec % (this.FixInterval * 1000)
		if v != 0 {
			return false
		} else {
			return true
		}
	}
	return true
}

func (this *GnssRinexObs3xWriterRt) checkFilterNaviSys(obs *gnss.RtcmMsm) bool {
	return true
}

func (this *GnssRinexObs3xWriterRt) WriteDelaySecs() float64 {
	if this.__writing_flag == 0 {
		return -1
	}

	return time.Now().Sub(this.lastWriteCacheT).Seconds()
}

func (this *GnssRinexObs3xWriterRt) Reset() {

}

func (this *GnssRinexObs3xWriterRt) checkRinexHeader(matchfilename string) (rval *gnss.RinexHeader) {
	lst := this.cacheEpochReclst
	if lst == nil {
		return nil
	}

	j := 0
	lst.Range(func(args ...interface{}) bool {
		rec := args[0].(*gnssRinexObs3xWriterEpochCacheRec)
		if len(matchfilename) == 0 {
			matchfilename = rec.ObsFileName
		}
		if rec.ObsFileName == matchfilename {
			rec.epochList.Range(func(idx int, value interface{}, removeit *bool) bool {
				epoch := value.(*gnss.RtcmMsmEpoch)
				epoch.Foreach(func(sys byte, msm *gnss.RtcmMsm) {
					if rval == nil {
						rval = gnss.NewRinexHeader()
					}
					rval.CheckAddMsmHeader(msm.Header)
				})
				j++
				return true
			})
			if j >= 60 { // 60个历元
				return false // 不需要再循环
			}
		}
		return true
	})

	return rval
}

func (this *GnssRinexObs3xWriterRt) OnRecvGnssDataRec(sender interface{}, rec *gnss.GnssRtcmData) {

	if rec.TypeId == gnss.TYPE_OBS {
		if obs, ok := rec.Data.(*gnss.RtcmMsm); ok {
			if this.msmInput == nil {
				this.msmInput = gnss.NewRtcmMsmEpochInput()

				this.msmInput.OnRtcmMsmEpochEnd = func(epoch *gnss.RtcmMsmEpoch) { // 收到一个历元

					this.checkInputEpoch(epoch)

					if this.WriteInterval > 0 {
						t1 := this.lastWriteCacheT
						if t1.IsZero() {
							t1 = this.firstRecvRtcmT
						}
						if time.Now().Sub(t1) >= this.WriteInterval { // 需要进行写入?
							this.checkPostCurrentCache() // 进缓存
							this.requestWrite2File()
						}
					}

				}
			}

			if this.firstRecvRtcmT.IsZero() {
				this.firstRecvRtcmT = time.Now()
			}

			if obs.MsmTime.IsZero() {
				obs.MsmTime = rec.RtcmTime
			}

			//obs.MsmTime = rec.RtcmTime
			//obs.ProcObs()    // 最好不处理 交给调用者执行

			this.msmInput.InputMsm(obs, func() bool {
				return this.checkFilterNaviSys(obs) && this.checkFilterMsm(obs)
			})
		}
	}
}
